/**
 * Copyright 2014 wasu.com
 *
 *
 * Create on 2015-1-9 下午11:28:33
 */
package xj.toolkit.jms;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.jms.ConnectionFactory;
import javax.jms.MessageListener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.SimpleMessageListenerContainer;

import com.google.common.collect.Lists;

/**
 * topic/queue:target.name, -- 目标名称:topic or queue
 * 设置需要监听的消息队列名称，创建Listener，创建一个MessageListenerContainer，并且管理他们。
 * 
 * @author <a href="mailto:caoxiaojian@wasu.com>xiaojian.cao</a>
 * 
 */
public abstract class MessageLisetenerContainerManager {

	private static final Logger logger = LoggerFactory
			.getLogger(MessageLisetenerContainerManager.class);

	private ConnectionFactory connectionFactory = null;

	private int executorPoolSizeLimit = 10;

	private ThreadPoolExecutor workService = null;

	private int concurrentConsumers = 1;

	/**
	 * 所有被监听的目标队列名称。
	 */
	private List<String> listenTargetNames = Lists.newArrayList();

	private List<AbstractMessageListenerContainer> messageListenerContainers = Lists
			.newArrayList();

	/**
	 * 得到一个消息监听器。
	 * 
	 * @return 消息监听器。
	 */
	protected abstract MessageListener obtainMessageListener();

	/**
	 * @return the listenTargetNames
	 */
	public List<String> getListenTargetNames() {
		return Lists.newArrayList(listenTargetNames);
	}

	/**
	 * @param listenTargetNames
	 *            the listenTargetNames to set
	 */
	public void setListenTargetNames(List<String> listenTargetNames) {
		this.listenTargetNames = listenTargetNames;
	}

	/**
	 * @param connectionFactory
	 *            the connectionFactory to set
	 */
	public void setConnectionFactory(ConnectionFactory connectionFactory) {
		this.connectionFactory = connectionFactory;
	}

	/**
	 * @param executorPoolSizeLimit
	 *            the executorPoolSizeLimit to set
	 */
	public void setExecutorPoolSizeLimit(int executorPoolSizeLimit) {
		this.executorPoolSizeLimit = executorPoolSizeLimit;
	}

	/**
	 * @param concurrentConsumers
	 *            the concurrentConsumers to set
	 */
	public void setConcurrentConsumers(int concurrentConsumers) {
		this.concurrentConsumers = concurrentConsumers;
	}

	public void start() {
		if (listenTargetNames == null || listenTargetNames.size() == 0) {
			logger.warn("NULL or EMPTY targets?");
			return;
		}

		this.workService = new ThreadPoolExecutor(0, executorPoolSizeLimit, 1L,
				TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
				new ThreadFactory() {
					public Thread newThread(Runnable r) {
						return new Thread(r, "JMS-MESSAGE-LISTENER");
					}
				});

		for (String listenTargetName : listenTargetNames) {
			String[] str = listenTargetName.split(":");
			if (str.length != 2) {
				logger.error("Invalid format[" + listenTargetName
						+ "], right[topic/queue:target.name].");
				continue;
			}

			boolean topic = str[0].equals("topic");
			String targetName = str[1];

			SimpleMessageListenerContainer contanier = new SimpleMessageListenerContainer();
			contanier.setPubSubDomain(topic);
			contanier.setDestinationName(targetName);
			contanier.setConnectionFactory(connectionFactory);

			contanier.setMessageListener(obtainMessageListener());
			
			if (this.workService != null) {
				contanier.setTaskExecutor(workService);
			}
			
			if (this.concurrentConsumers > 1) {
				contanier.setConcurrentConsumers(concurrentConsumers);
			}

			contanier.afterPropertiesSet();
			contanier.start();

			messageListenerContainers.add(contanier);

			logger.debug("New message listener container with["
					+ listenTargetName + "].");
		}
	}

	public void destory() {
		
		this.workService.shutdown();
		
		for (AbstractMessageListenerContainer container : messageListenerContainers) {
			container.stop();
			container.destroy();
		}
		
	}
}
